package com.lry;

import com.lry.config.FlushDiskType;
import com.lry.lock.PutMessageLock;
import com.lry.lock.PutMessageReentrantLock;
import com.lry.lock.PutMessageSpinLock;
import com.lry.message.MessageDecoder;
import com.lry.message.MessageExt;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Data
public class CommitLog {

    public final static int MESSAGE_MAGIC_CODE = -626843481;

    private final MappedFileQueue mappedFileQueue;

    private final DefaultMessageStore defaultMessageStore;

    protected HashMap<String/* topic-queueid */, Long/* offset */> topicQueueTable = new HashMap<String, Long>(1024);

    protected volatile long confirmOffset = -1L;

    protected final PutMessageLock putMessageLock;

    private volatile long beginTimeInLock = 0;

    private final FlushCommitLogService flushCommitLogService;//flush线程

    private final FlushCommitLogService commitLogService;//commit 线程

    private long reputFromOffset;

    public CommitLog(final DefaultMessageStore defaultMessageStore) {
        String storePath = defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog();
        this.mappedFileQueue = new MappedFileQueue(storePath,
                defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(), new MappedFileQueue.MappedFileCallback() {
            @Override
            public int calWritePosition(MappedFile mappedFile) {
                return CommitLog.this.calWritePosition(mappedFile);
            }
        });

        this.defaultMessageStore = defaultMessageStore;

        if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            this.flushCommitLogService = new GroupCommitService();
        } else {
            this.flushCommitLogService = new FlushRealTimeService();
        }

        this.commitLogService = new CommitRealTimeService();

        this.putMessageLock = defaultMessageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
    }

    //4(storeSize) +
    public int calWritePosition(MappedFile mappedFile){
        MappedByteBuffer mappedByteBuffer = mappedFile.getMappedByteBuffer();
        ByteBuffer slice = mappedByteBuffer.slice();
        slice.position(0);

        int writePosition = 0;
        int storeSize = -1;

        while(storeSize!=0 && slice.hasRemaining()){
            storeSize = slice.getInt();
            if(storeSize!=0){
                byte[] bs = new byte[storeSize-4];
                slice.get(bs);
                writePosition += storeSize;
            }
        }
        //可能有问题，因为在磁盘的commit log不一定reput到了 consume queue和 index file
        this.reputFromOffset = writePosition;
        return writePosition;
    }

    public void start(){
        this.flushCommitLogService.start();
        if (defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            this.commitLogService.start();
        }
    }

    public void shutdown() {
        if (defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            this.commitLogService.shutdown();
        }

        this.flushCommitLogService.shutdown();
    }

    public boolean load() {
        boolean result = this.mappedFileQueue.load();
        log.info("load commit log " + (result ? "OK" : "Failed"));
        return result;
    }

    public long flush() {
        this.mappedFileQueue.commit(0);
        this.mappedFileQueue.flush(0);
        return this.mappedFileQueue.getFlushedWhere();
    }
    public long getMinOffset() {
        MappedFile mappedFile = this.mappedFileQueue.getFirstMappedFile();
        if (mappedFile != null) {
            return mappedFile.getFileFromOffset();
        }

        return -1;
    }

    public long getMaxOffset() {
        return this.mappedFileQueue.getMaxOffset();
    }

    public long remainHowManyDataToCommit() {
        return this.mappedFileQueue.remainHowManyDataToCommit();
    }

    public long remainHowManyDataToFlush() {
        return this.mappedFileQueue.remainHowManyDataToFlush();
    }

    public SelectMappedBufferResult select(final long offset,final int size){
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset==0);
        if(null!=mappedFile){
            return mappedFile.select((int) (offset % defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog()),size);
        }
        return null;
    }
    public SelectMappedBufferResult select(final long offset){
        return select(offset,offset==0);
    }

    public SelectMappedBufferResult select(final long offset,boolean returnFirstOnNotFound){
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, returnFirstOnNotFound);
        if(null!=mappedFile){
            return mappedFile.select((int) (offset % defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog()));
        }
        return null;
    }



    //落盘，ha，
    public CompletableFuture<PutMessageResult> asyncPutMsg(final MessageExt msg){
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
        msg.setStoreTimestamp(beginLockTimestamp);

        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        this.beginTimeInLock = this.defaultMessageStore.getSystemClock().now();
        try {
            MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
                log.error("create mapped file1 error, topic: " + msg.getTopic());
                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED));
            }
            msg.setCommitLogOffset(mappedFile.calCurCommitLogOffset());
            boolean flag = mappedFile.append(MessageDecoder.encode(msg,true));
            if(!flag){
                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR));
            }
        } catch (Exception e) {
            e.printStackTrace();
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR));
        } finally {
            this.beginTimeInLock = 0;
            putMessageLock.unlock();
        }
        long elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;

        if (elapsedTimeInLock > 500) {
            log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}", elapsedTimeInLock);
        }

        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK);

        CompletableFuture<PutMessageStatus> flushOKFuture = submitFlushRequest(msg);

        return flushOKFuture.thenApply(flushStatus->{
            if (flushStatus != PutMessageStatus.PUT_OK) {
                putMessageResult.setPutMessageStatus(flushStatus);
            }
            return putMessageResult;
        });
    }

    public CompletableFuture<PutMessageStatus> submitFlushRequest(MessageExt messageExt) {
        // Synchronization flush
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
            if (messageExt.isWaitStoreMsgOK()) {
                GroupCommitRequest request = new GroupCommitRequest(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
                service.putRequest(request);
                return request.future();
            } else {
                service.wakeup();
                return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
            }
        }
        // Asynchronous flush
        else {
            if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                flushCommitLogService.wakeup();
            } else  {
                commitLogService.wakeup();
            }
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }

    public void destroy() {
    }

    public DispatchRequest populateDispatchRequest(ByteBuffer byteBuffer) {
        MessageExt decode = MessageDecoder.decode(byteBuffer);

        return new DispatchRequest(decode.getTopic(),decode.getQueueId(),decode.getCommitLogOffset(),
                decode.getStoreSize(), StringUtils.isEmpty(decode.getTags())?0:decode.getTags().hashCode(),decode.getStoreTimestamp(),
                decode.getQueueOffset(),decode.getKeys(),decode.getProperties());
    }

    public static class GroupCommitRequest {
        private CompletableFuture<PutMessageStatus> flushOKFuture = new CompletableFuture<>();
        private final long deadLine;

        public GroupCommitRequest(long timeoutMillis) {
            this.deadLine = System.nanoTime() + (timeoutMillis * 1_000_000);
        }

        public long getDeadLine() {
            return deadLine;
        }


        public void wakeupCustomer(final PutMessageStatus putMessageStatus) {
            this.flushOKFuture.complete(putMessageStatus);
        }

        public CompletableFuture<PutMessageStatus> future() {
            return flushOKFuture;
        }

    }

    abstract class FlushCommitLogService extends ServiceThread {
        protected static final int RETRY_TIMES_OVER = 10;
    }

    class GroupCommitService extends FlushCommitLogService {

        private volatile LinkedList<GroupCommitRequest> requestsWrite = new LinkedList<GroupCommitRequest>();
        private volatile LinkedList<GroupCommitRequest> requestsRead = new LinkedList<GroupCommitRequest>();
        private final PutMessageSpinLock lock = new PutMessageSpinLock();

        @Override
        public String getServiceName() {
            return GroupCommitService.class.getSimpleName();
        }

        @Override
        public long getJointime() {
            return 1000 * 60 * 5;
        }

        @Override
        protected void onWaitEnd() {
            this.swapRequests();
        }
        @Override
        public void run() {
            CommitLog.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.waitForRunning(10);
                    this.doCommit();
                } catch (Exception e) {
                    CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            // Under normal circumstances shutdown, wait for the arrival of the
            // request, and then flush
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                CommitLog.log.warn(this.getServiceName() + " Exception, ", e);
            }

            synchronized (this) {
                this.swapRequests();
            }

            this.doCommit();

            CommitLog.log.info(this.getServiceName() + " service end");
        }
        private void doCommit() {
            if (!this.requestsRead.isEmpty()) {

                for (GroupCommitRequest req : this.requestsRead) {

                    CommitLog.this.mappedFileQueue.flush(0);

                    req.wakeupCustomer(PutMessageStatus.PUT_OK);
                }

                this.requestsRead = new LinkedList<>();
            } else {
                // Because of individual messages is set to not sync flush, it
                // will come to this process
                CommitLog.this.mappedFileQueue.flush(0);
            }
        }
        private void swapRequests() {
            lock.lock();
            try {
                LinkedList<GroupCommitRequest> tmp = this.requestsWrite;
                this.requestsWrite = this.requestsRead;
                this.requestsRead = tmp;
            } finally {
                lock.unlock();
            }
        }

        public void putRequest(GroupCommitRequest request) {
            lock.lock();
            try {
                this.requestsWrite.add(request);
            } finally {
                lock.unlock();
            }
            this.wakeup();
        }
    }
    class CommitRealTimeService extends FlushCommitLogService {


        @Override
        public String getServiceName() {
            return CommitRealTimeService.class.getSimpleName();
        }

        @Override
        public void run() {

        }
    }

    //
    class FlushRealTimeService extends FlushCommitLogService {

        @Override
        public String getServiceName() {
            return FlushRealTimeService.class.getSimpleName();
        }

        @Override
        public void run() {

        }
    }
}
